package com.abyss.transformation.datastream;


import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

/**
 * 通过Split对流进行拆分
 * 通过Select对拆分后的流进行选择
 */
public class SplitAndSelectDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. source
        DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4, 5, 6);

        // 3. split 来区分偶数和奇数
        SplitStream<Integer> splitStream = source.split(new OutputSelector<Integer>() {
            @Override
            public Iterable<String> select(Integer value) {
                if (value % 2 == 0) {
                    return Arrays.asList("even", "偶数", "余2得0");
                } else {
                    return Arrays.asList("odd", "奇数", "余2得1");
                }
            }
        });

        DataStream<Integer> odd = splitStream.select("奇数");
        DataStream<Integer> even = splitStream.select("余2得0");

        odd.print("odd>>>");
        even.print("even>>>");

        env.execute();
    }
}